/**
 * Copyright (c) 2019, ShangHai YTO Co., Ltd.
 * All right reserved.
 *
 * THIS IS UNPUBLISHED PROPRIETARY SOURCE CODE OF YTO
 * CO., LTD.  THE CONTENTS OF THIS FILE MAY NOT BE DISCLOSED
 * TO THIRD PARTIES, COPIED OR DUPLICATED IN ANY FORM, IN WHOLE OR IN PART,
 * WITHOUT THE PRIOR WRITTEN PERMISSION OF YTO
 * CO., LTD.
 */
package org.sc.xl.guavatest.doris;

import java.sql.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @description: TODO 请在此添加描述
 * @reason: TODO ADD REASON(可选)
 * @author FXL
 *
 * @date 2023/9/5 16:31 
 * @since JDK 1.8
 */
public class MysqlToDoris {


    static final String mysqlUrl = "jdbc:mysql://10.130.253.67:3306/bigdata_mutiple?useUnicode=true&characterEncoding=UTF-8&useSSL=false";

    static final String mysqlUsername = "root";

    static final String mysqlPassword = "0JZ^YQHxjII!p1Qf";

    static final String dorisUrl = "jdbc:mysql://10.130.253.66:9030/ytrpt?rewriteBatchedStatements=True";

    static final String dorisUsername = "root";

    static final String dorisPassword = null;

    static final String select = "SELECT * FROM ";

    static final String tableName = "t_app_major_market_d";

    static final int numThreads = 5;

    static final int batchSize = 3000;


    public static void main(String[] args) throws SQLException {
        PreparedStatement preparedStatement = null;
        Connection mysqlConnection = null;
        Connection dorisConnection = null;
        ResultSet mysqlResultSet = null;
        Statement statement = null;
        ExecutorService executorService = null;

        try {
            mysqlConnection = DriverManager.getConnection(mysqlUrl, mysqlUsername, mysqlPassword);
            statement = mysqlConnection.createStatement();

            StringBuilder mysqlQuery = new StringBuilder();
            mysqlQuery.append(select).append(tableName);

            mysqlResultSet = statement.executeQuery(mysqlQuery.toString());
            executorService = Executors.newFixedThreadPool(numThreads);

            dorisConnection = DriverManager.getConnection(dorisUrl, dorisUsername, dorisPassword);


            int counter = 0;

            while (mysqlResultSet.next()) {

                if (preparedStatement == null) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append("insert into ").append(tableName).append(" (");

                    ResultSetMetaData mysqlMetaData = mysqlResultSet.getMetaData();

                    int columnCount = mysqlMetaData.getColumnCount();

                    for (int i = 1; i <= columnCount; i++) {
                        stringBuilder.append(mysqlMetaData.getColumnName(i));
                        if (i < columnCount) {
                            stringBuilder.append(", ");
                        }
                    }

                    stringBuilder.append(") values (");

                    for (int i = 1; i <= columnCount; i++) {
                        stringBuilder.append("?");
                        if (i < columnCount) {
                            stringBuilder.append(", ");
                        }
                    }

                    stringBuilder.append(")");
                    System.out.println(stringBuilder);
                    preparedStatement = dorisConnection.prepareStatement(stringBuilder.toString());
                }

                ResultSetMetaData mysqlMetaData = mysqlResultSet.getMetaData();
                int columnCount = mysqlMetaData.getColumnCount();
                for (int i = 1; i <= columnCount; i++) {
                    preparedStatement.setObject(i, mysqlResultSet.getObject(i));
                }
                preparedStatement.addBatch();
                counter++;
                if (counter % batchSize == 0) {
                    PreparedStatement finalPreparedStatement = preparedStatement;
                    Connection finalDorisConnection = dorisConnection;
                    executorService.submit(() -> {
                        try {
                            finalPreparedStatement.executeBatch();
                            finalDorisConnection.setAutoCommit(true);
                            finalDorisConnection.commit();
                        } catch (SQLException e) {
                            throw new RuntimeException(e);
                        }
                    });
                    preparedStatement = null;
                }
            }
            if (preparedStatement != null) {
                preparedStatement.executeBatch();
                dorisConnection.setAutoCommit(false);
                dorisConnection.commit();
            }
            System.err.println("finish!");
        } catch (SQLException e) {
            throw new RuntimeException(e);
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
            if (dorisConnection != null) {
                dorisConnection.close();
            }
            mysqlResultSet.close();
            statement.close();
            mysqlConnection.close();

            executorService.shutdown();
        }
    }
}